package com.atguigu.bigdata.edu.realtime.app.dwd;

import com.atguigu.bigdata.edu.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


import java.time.Duration;

public class Dwd_06_DwdExamPaperDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_06_DwdExamPaperDetail().init(
                3006,
                2,
                "Dwd_06_DwdExamPaperDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //从ods_db读取数据
        readOdsDb(tEnv, "Dwd_06_DwdExamPaperDetail");

        //sql 方式 过滤考试主题 试卷粒度表
        Table testExamInfo = tEnv.sqlQuery(" select " +
                " data['id'] id," +
                " cast(data['paper_id'] as bigint) paper_id," +
                " data['user_id'] user_id," +
                " data['score'] score," +
                " data['duration_sec'] duration_sec," +
                " data['submit_time'] submit_time," +
                " ts," +
                " pt" +
                " from ods_db" +
                " where `database`='edu' and" +
                " `table`='test_exam'" );
        tEnv.createTemporaryView("test_exam",testExamInfo);
        // 从mysql直接读取出试卷表 test_paper
        tEnv.executeSql("create table test_paper( " +
                " id bigint," +
                " paper_title string," +
                " course_id bigint" +
                " )" +
                "WITH (" +
                "    'connector' = 'jdbc'," +
                "    'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "    'url' = 'jdbc:mysql://hadoop102:3306/edu?useSSL=false'," +
                "    'table-name' = 'test_paper'," +
                "    'username' = 'root', " +
                "    'password' = 'aaaaaa'," +
                "    'lookup.cache.max-rows' = '10'," +
                "    'lookup.cache.ttl' = '1 hour' " +
                "  )");
        tEnv.sqlQuery("select * from test_paper ").execute().print();
        //从mysql 过滤出 课程表
        tEnv.executeSql("create table course_info(" +
                " id bigint," +
                " course_name string" +
                ") with (" +
                " 'connector'='jdbc'," +
                " 'driver'='com.mysql.cj.jdbc.Driver'," +
                " 'url' = 'jdbc:mysql://hadoop102:3306/edu?useSSL=false'," +
                " 'table-name' = 'course_info'," +
                " 'username' = 'root'," +
                " 'password' = 'aaaaaa'," +
                " 'lookup.cache.max-rows' = '10'," +
                " 'lookup.cache.ttl' = '1 hour' " +
                ")");
        //tEnv.sqlQuery("select * from course_info ").execute().print();
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        //将ods_db的表 与test_paper进行 look up join
        Table sqlQuery = tEnv.sqlQuery("select " +
                " te.id id," +
                " te.paper_id paper_id," +
                " tp.paper_title paper_title," +
                " te.user_id user_id," +
                " tp.course_id course_id," +
                " te.score score," +
                " te.duration_sec duration_sec," +
                " te.submit_time submit_time," +
                " te.pt pt," +
                " te.ts ts" +
                " from test_exam te" +
                " join test_paper for system_time as of te.pt as tp on te.paper_id=tp.id");
        tEnv.createTemporaryView("t",sqlQuery);

        // 将上表与course_info 进行look up join
        Table result = tEnv.sqlQuery("select " +
                " t.id," +
                " t.paper_id," +
                " t.paper_title," +
                " t.user_id," +
                " t.course_id," +
                " ci.course_name," +
                " t.score," +
                " t.duration_sec," +
                " t.submit_time" +
                " from t " +
                " join course_info for system_time as of t.pt as ci on t.course_id=ci.id");
        //result.execute().print();

        //写出到 kafka  在kafka创建表
        tEnv.executeSql("create table dwd_exam_paper_detail(" +
                " id string," +
                " paper_id bigint," +
                " paper_title string," +
                " user_id string," +
                " course_id bigint," +
                " course_name string ," +
                " score string," +
                " duration_sec string," +
                " submit_time string," +
                " primary key(id) not enforced  " +
                ")" + SQLUtil.getUpsertKafkaSink(Constant.DWD_EXAM_PAPER_DETAIL));

        result.executeInsert("dwd_exam_paper_detail");
    }


}
